import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClientExample {

    public static void main(String[] args) {
        // 连接到 Canal Server
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("localhost", 11111), // Canal Server 的地址和端口
                "example",  // destination
                "",  // username
                ""   // password
        );

        try {
            // 连接到 Canal
            connector.connect();
            connector.subscribe(".*\\..*"); // 订阅所有数据库和表
            connector.rollback();

            while (true) {
                // 获取指定数量的消息
                Message message = connector.getWithoutAck(1000);
                long batchId = message.getId();
                int size = message.getEntries().size();

                if (batchId != -1 && size > 0) {
                    // 处理数据
                    printEntries(message.getEntries());
                }

                // 确认消息处理完成
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                try {
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        if (rowChange.getEventType() == CanalEntry.EventType.INSERT) {
                            System.out.println("INSERT: " + rowData.getAfterColumnsList());
                        } else if (rowChange.getEventType() == CanalEntry.EventType.UPDATE) {
                            System.out.println("UPDATE: " + rowData.getAfterColumnsList());
                        } else if (rowChange.getEventType() == CanalEntry.EventType.DELETE) {
                            System.out.println("DELETE: " + rowData.getBeforeColumnsList());
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
